#include <network\streamingConnection.h>

#include <commands\commandFactory.h>
#include <commands\command.h>

#include <app\states\cacheState.h>

#include <iostream>

#undef DEBUG_INFO

streamingConnection::streamingConnection(boost::asio::io_service& io_service) : socket_(io_service), io_service_(io_service) {
  input_bytes_count = 0;
  input_command_size = 0;
  write_in_progress = false;
  input_bytes = new stringstream(ios_base::in | ios_base::out | ios_base::binary);
}

session &
streamingConnection::getSession() {
  return my_session;
}

tcp::socket& 
streamingConnection::socket() {
  return socket_;
}

void
streamingConnection::start() {
  socket_.async_read_some(boost::asio::buffer(data_, max_length),
      boost::bind(&streamingConnection::handle_read, this,
        boost::asio::placeholders::error,
        boost::asio::placeholders::bytes_transferred));
  #ifdef DEBUG_INFO
  cerr << "starting new connection" << endl;
  #endif
}

void
streamingConnection::receiveBytes(size_t bytes_transferred) {

  // append the transfered data to input stream and count bytes
  if (bytes_transferred != 0) {
    input_bytes->write(data_, (std::streamsize) bytes_transferred);
    input_bytes_count += (unsigned int) bytes_transferred;
  }

  // check if we expect new command and is available size of the command in bytes
  if (input_command_size == 0 && input_bytes_count >= sizeof(int)) {
    input_bytes->seekg(0, ios::beg);
    input_bytes->read((char *) &input_command_size, sizeof(unsigned int));
  }

  #ifdef DEBUG_INFO
  std::cerr << "input bytes count/command size =" << input_bytes_count << "/" << input_command_size << std::endl;
  #endif

  // create the command when sufficient data are available
  if (input_bytes_count >= input_command_size) {

    // create command and consume bytes for command
    input_bytes->seekg(0, ios::beg);

    // skip data size bytes
    input_bytes->seekg(sizeof(unsigned int), ios::beg);
    // read command id
    unsigned int commandId = 0;
    input_bytes->read((char *) &commandId, sizeof(unsigned int));

    #ifdef DEBUG_INFO
    std::cerr << "CREATE command with id: " << commandId << std::endl;
    #endif

    // create command
    if (commandId != 0) {
      command *cmd = commandFactory::instance().createCommand(commandId, this);
      if (cmd) {
        // the decoded data size is input_command_size - cmd data size and cmd id
        cmd->decode(*input_bytes, input_command_size - 2 * sizeof(unsigned int));
        delete cmd;
      }
    }

    if ((input_bytes_count - input_command_size) != 0) {
      
      unsigned char *rest_bytes = new unsigned char[input_bytes_count - input_command_size];
      input_bytes->seekg(input_command_size, ios::beg);
      input_bytes->read((char *) rest_bytes, input_bytes_count - input_command_size);

      // keep the rest in input stream
      delete input_bytes;
      input_bytes = new stringstream(ios_base::in | ios_base::out | ios_base::binary);
      input_bytes->write((char *) rest_bytes, input_bytes_count - input_command_size);
     
      delete [] rest_bytes;
    }
    else {
      delete input_bytes;
      input_bytes = new stringstream(ios_base::in | ios_base::out | ios_base::binary);
    }

    input_bytes_count = input_bytes_count - input_command_size;
    input_command_size = 0;
    // perform the rest in case there are simultaneous requests e.g. input_bytes_count > input_command_size
    if ((input_bytes_count - input_command_size) != 0) receiveBytes(0);
  }
}

void 
streamingConnection::handle_read(const boost::system::error_code& error, size_t bytes_transferred) {
  if (!error) {

    #ifdef DEBUG_INFO
    std::cerr << "HANDLEd " << bytes_transferred << " input bytes" << std::endl;
    #endif
    receiveBytes(bytes_transferred);

    socket_.async_read_some(boost::asio::buffer(data_, max_length),
        boost::bind(&streamingConnection::handle_read, this,
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred));
  }
  else {
    socket_.close();
    delete this;
  }
}


serialize::serialize() {
  current_ticket = 0;
  last_ticket = 0;
}

int
serialize::getTicket() {
  int new_ticket;
  
  input_mutex.lock();
    new_ticket = last_ticket + 1;
    if (last_ticket >= MAX_TICKETS) new_ticket = 0;
  input_mutex.unlock();

  return new_ticket;
}

void 
serialize::relaseTicket(int ticket) {
  input_mutex.lock();
    current_ticket++;
    if (current_ticket >= MAX_TICKETS) current_ticket = 0;
  input_mutex.unlock();
}

void 
streamingConnection::sendData(unsigned char *bytes, int size) {
  
  output_mutex.lock();
    output_bytes.push_back(bytes);
    output_bytes_size.push_back(size);
  output_mutex.unlock();

  io_service_.post(boost::bind(&streamingConnection::writeBytes, this, bytes, size));
}

void 
streamingConnection::writeBytes(unsigned char *bytes, int size) {

  output_mutex.lock();
  
  if (!write_in_progress && output_bytes.size() != 0 && output_bytes_size.size() != 0) {
    write_in_progress = true;
    unsigned char *_output_bytes    = output_bytes[0];
    unsigned int _output_bytes_size = output_bytes_size[0];

    boost::asio::async_write(socket_,
          boost::asio::buffer(_output_bytes, _output_bytes_size),
          boost::bind(&streamingConnection::handle_write, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
  }
  output_mutex.unlock();
}

void 
streamingConnection::handle_write(const boost::system::error_code& error, size_t bytes_transferred) {
  if (!error) {

    output_mutex.lock();

      //unsigned char *_output_bytes    = output_bytes[0];
      //unsigned int _output_bytes_size = output_bytes_size[0];

      #ifdef DEBUG_INFO
      //std::cerr << "written " << bytes_transferred << " bytes" << std::endl;
      #endif

      if (bytes_transferred == output_bytes_size[0]) {
        #ifdef DEBUG_INFO
        //std::cerr << "END write all " << output_bytes_size[0] << " bytes" << std::endl;
        #endif
        unsigned char * _output_bytes    = output_bytes[0];
        output_bytes.erase(output_bytes.begin());
        output_bytes_size.erase(output_bytes_size.begin());
        delete [] _output_bytes;
      }

      if (!output_bytes.empty()) {
        write_in_progress = true;
        boost::asio::async_write(socket_,
            boost::asio::buffer(output_bytes[0], output_bytes_size[0]),
            boost::bind(&streamingConnection::handle_write, this,
              boost::asio::placeholders::error,
              boost::asio::placeholders::bytes_transferred));
      }
      else {
        write_in_progress = false;
      }

    output_mutex.unlock();
  }
  else {
    socket_.close();
    delete this;
  }
}
